package com.dtruth.dataflow.bolt;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dtruth.dataflow.dao.HBaseDAO;
import com.dtruth.dataflow.util.MyLoggerFactory;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class Save2HBaseBolt extends BaseBasicBolt {

	private static Logger log = MyLoggerFactory.getLogger(Save2HBaseBolt.class);
	private static final long serialVersionUID = 1L;

	HBaseDAO hdao;

	/**
	 * On create
	 */
	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		hdao = new HBaseDAO();
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		@SuppressWarnings("unchecked")
		Map<String, Object> map = (Map<String, Object>) input.getValue(0);
		map.put("updatetime", System.currentTimeMillis());
		save2Hbase(map);
	}

	private void save2Hbase(Map<String, Object> map) {
		String rowKey = map.get("key").toString();
		for (Map.Entry<String, Object> entry : map.entrySet()) {
			if (!entry.getKey().equals("key")) {
				String qulifier = entry.getKey();
				String value = entry.getValue() == null ? "" : entry.getValue()
						.toString();
				if (!value.equals(""))
					hdao.writeRow("news", rowKey, "data", qulifier, value);
			}
		}

	}

}